/******************************************************
*  Copyright(c) pages.ilovezxl.top
*  FileName :thread_linux.c
*  Author : rivulet.dave
*  Description : 线程同步LINUX实现
*  Data : 2020-07-10
******************************************************/

/*********************************************************************
 * INCLUDES
 */


#include "stdrope.h"

#include "sallib.h"
#include "salagent.h"
#include "link.h"

#if (SAL_PRODUCER == _SAL_LINUX_)
#include <pthread.h>
#include <unistd.h>
#include <sys/types.h>
#include <sys/time.h>
#include <semaphore.h>
#include <mqueue.h>
#include <errno.h>

#include "mem.h"
#include "threadasync.h"
#include "threadasync_linux.h"

/*********************************************************************
 *  EXTERNAL VARIABLES
 */

/*********************************************************************
 * MACROS
 */


/*********************************************************************
 * TYPES
 */

typedef struct _mutex_linux
{
    const char *pName;
    osThreadId_t pOwner;
    pthread_mutex_t stMutex;
    pthread_mutexattr_t stMutexAttr;
} MUTEX_LINUX_CB_S;

typedef struct _semaphore_linux
{
    const char *pName;
    sem_t stSem;
} SEM_LINUX_CB_S;

typedef struct _msg_linux
{
    const char *pName;
    uint32_t uiMsgSize;
    uint32_t uiMsgCnt;
    mqd_t iMsgQCmd;
} MSG_LINUX_CB_S;

/*********************************************************************
 * FUNCTIONS DECLARE
 */

static osEventFlagsId_t osEventFlagsNewByLinux(const osEventFlagsAttr_t *attr);
static const char *osEventFlagsGetNameByLinux(osEventFlagsId_t ef_id);
static uint32_t osEventFlagsSetByLinux(osEventFlagsId_t ef_id, uint32_t flags);
static uint32_t osEventFlagsClearByLinux(osEventFlagsId_t ef_id, uint32_t flags);
static uint32_t osEventFlagsGetByLinux(osEventFlagsId_t ef_id);
static uint32_t osEventFlagsWaitByLinux(osEventFlagsId_t ef_id, uint32_t flags, uint32_t options, uint32_t timeout);
static osStatus_t osEventFlagsDeleteByLinux(osEventFlagsId_t ef_id);
static osMutexId_t osMutexNewByLinux(const osMutexAttr_t *attr);
static const char *osMutexGetNameByLinux(osMutexId_t mutex_id);
static osStatus_t osMutexAcquireByLinux(osMutexId_t mutex_id, uint32_t timeout);
static osStatus_t osMutexReleaseByLinux(osMutexId_t mutex_id);
static osThreadId_t osMutexGetOwnerByLinux(osMutexId_t mutex_id);
static osStatus_t osMutexDeleteByLinux(osMutexId_t mutex_id);
static osSemaphoreId_t osSemaphoreNewByLinux(uint32_t max_count, uint32_t initial_count, const osSemaphoreAttr_t *attr);
static const char *osSemaphoreGetNameByLinux(osSemaphoreId_t semaphore_id);
static osStatus_t osSemaphoreAcquireByLinux(osSemaphoreId_t semaphore_id, uint32_t timeout);
static osStatus_t osSemaphoreReleaseByLinux(osSemaphoreId_t semaphore_id);
static uint32_t osSemaphoreGetCountByLinux(osSemaphoreId_t semaphore_id);
static osStatus_t osSemaphoreDeleteByLinux(osSemaphoreId_t semaphore_id);
static osMessageQueueId_t osMessageQueueNewByLinux(uint32_t msg_count, uint32_t msg_size, const osMessageQueueAttr_t *attr);
static const char *osMessageQueueGetNameByLinux(osMessageQueueId_t mq_id);
static osStatus_t osMessageQueuePutByLinux(osMessageQueueId_t mq_id, const void *msg_ptr, uint8_t msg_prio, uint32_t timeout);
static osStatus_t osMessageQueueGetByLinux(osMessageQueueId_t mq_id, void *msg_ptr, uint8_t *msg_prio, uint32_t timeout);
static uint32_t osMessageQueueGetCapacityByLinux(osMessageQueueId_t mq_id);
static uint32_t osMessageQueueGetMsgSizeByLinux(osMessageQueueId_t mq_id);
static uint32_t osMessageQueueGetCountByLinux(osMessageQueueId_t mq_id);
static uint32_t osMessageQueueGetSpaceByLinux(osMessageQueueId_t mq_id);
static osStatus_t osMessageQueueResetByLinux(osMessageQueueId_t mq_id);
static osStatus_t osMessageQueueDeleteByLinux(osMessageQueueId_t mq_id);
static void osWaitTimerInitByLinux(struct timespec *pOutTime, uint32_t uiMs);

/*********************************************************************
 *  VARIABLES
 */

static OS_THREADASYNC_BASE_S s_stOSThreadAsyncLinuxObj;

/*********************************************************************
 * FUNCTIONS
 */

/**
 *   @Description: 获取线程同步对象
 *   @param : NA
 *   @return: 对象地址
 */
void *osThreadAsyncLinuxObjNew(void)
{
    memset(&s_stOSThreadAsyncLinuxObj, 0, sizeof(OS_THREADASYNC_BASE_S));
    
    s_stOSThreadAsyncLinuxObj.pfOSThreadFlagsSet = NULL;
    s_stOSThreadAsyncLinuxObj.pfOSThreadFlagsClear = NULL;
    s_stOSThreadAsyncLinuxObj.pfOSThreadFlagsGet = NULL;
    s_stOSThreadAsyncLinuxObj.pfOSThreadFlagsWait = NULL;
    s_stOSThreadAsyncLinuxObj.pfOSEventFlagsNew = osEventFlagsNewByLinux;
    s_stOSThreadAsyncLinuxObj.pfOSEventFlagsGetName = osEventFlagsGetNameByLinux;
    s_stOSThreadAsyncLinuxObj.pfOSEventFlagsSet = osEventFlagsSetByLinux;
    s_stOSThreadAsyncLinuxObj.pfOSEventFlagsClear = osEventFlagsClearByLinux;
    s_stOSThreadAsyncLinuxObj.pfOSEventFlagsGet = osEventFlagsGetByLinux;
    s_stOSThreadAsyncLinuxObj.pfOSEventFlagsWait = osEventFlagsWaitByLinux;
    s_stOSThreadAsyncLinuxObj.pfOSEventFlagsDelete = osEventFlagsDeleteByLinux;
    s_stOSThreadAsyncLinuxObj.pfOSMutexNew = osMutexNewByLinux;
    s_stOSThreadAsyncLinuxObj.pfOSMutexGetName = osMutexGetNameByLinux;
    s_stOSThreadAsyncLinuxObj.pfOSMutexAcquire = osMutexAcquireByLinux;
    s_stOSThreadAsyncLinuxObj.pfOSMutexRelease = osMutexReleaseByLinux;
    s_stOSThreadAsyncLinuxObj.pfOSMutexGetOwner = osMutexGetOwnerByLinux;
    s_stOSThreadAsyncLinuxObj.pfOSMutexDelete = osMutexDeleteByLinux;
    s_stOSThreadAsyncLinuxObj.pfOSSemaphoreNew = osSemaphoreNewByLinux;
    s_stOSThreadAsyncLinuxObj.pfOSSemaphoreGetName = osSemaphoreGetNameByLinux;
    s_stOSThreadAsyncLinuxObj.pfOSSemaphoreAcquire = osSemaphoreAcquireByLinux;
    s_stOSThreadAsyncLinuxObj.pfOSSemaphoreRelease = osSemaphoreReleaseByLinux;
    s_stOSThreadAsyncLinuxObj.pfOSSemaphoreGetCount = osSemaphoreGetCountByLinux;
    s_stOSThreadAsyncLinuxObj.pfOSSemaphoreDelete = osSemaphoreDeleteByLinux;
    s_stOSThreadAsyncLinuxObj.pfOSMessageQueueNew = osMessageQueueNewByLinux;
    s_stOSThreadAsyncLinuxObj.pfOSMessageQueueGetName = osMessageQueueGetNameByLinux;
    s_stOSThreadAsyncLinuxObj.pfOSMessageQueuePut = osMessageQueuePutByLinux;
    s_stOSThreadAsyncLinuxObj.pfOSMessageQueueGet = osMessageQueueGetByLinux;
    s_stOSThreadAsyncLinuxObj.pfOSMessageQueueGetCapacity = osMessageQueueGetCapacityByLinux;
    s_stOSThreadAsyncLinuxObj.pfOSMessageQueueGetMsgSize = osMessageQueueGetMsgSizeByLinux;
    s_stOSThreadAsyncLinuxObj.pfOSMessageQueueGetCount = osMessageQueueGetCountByLinux;
    s_stOSThreadAsyncLinuxObj.pfOSMessageQueueGetSpace = osMessageQueueGetSpaceByLinux;
    s_stOSThreadAsyncLinuxObj.pfOSMessageQueueReset = osMessageQueueResetByLinux;
    s_stOSThreadAsyncLinuxObj.pfOSMessageQueueDelete = osMessageQueueDeleteByLinux;

    return &s_stOSThreadAsyncLinuxObj;
}

//  ==== Event Flags Management Functions ====

/// Create and Initialize an Event Flags object.
/// \param[in]     attr          event flags attributes; NULL: default values.
/// \return event flags ID for reference by other functions or NULL in case of error.
static osEventFlagsId_t osEventFlagsNewByLinux (const osEventFlagsAttr_t *attr)
{
    return (osEventFlagsId_t)NULL;
}

/// Get name of an Event Flags object.
/// \param[in]     ef_id         event flags ID obtained by \ref osEventFlagsNew.
/// \return name as null-terminated string.
static const char *osEventFlagsGetNameByLinux (osEventFlagsId_t ef_id)
{
    return "";
}

/// Set the specified Event Flags.
/// \param[in]     ef_id         event flags ID obtained by \ref osEventFlagsNew.
/// \param[in]     flags         specifies the flags that shall be set.
/// \return event flags after setting or error code if highest bit set.
static uint32_t osEventFlagsSetByLinux (osEventFlagsId_t ef_id, uint32_t flags)
{
    return 0;
}

/// Clear the specified Event Flags.
/// \param[in]     ef_id         event flags ID obtained by \ref osEventFlagsNew.
/// \param[in]     flags         specifies the flags that shall be cleared.
/// \return event flags before clearing or error code if highest bit set.
static uint32_t osEventFlagsClearByLinux (osEventFlagsId_t ef_id, uint32_t flags)
{
    return 0;
}

/// Get the current Event Flags.
/// \param[in]     ef_id         event flags ID obtained by \ref osEventFlagsNew.
/// \return current event flags.
static uint32_t osEventFlagsGetByLinux (osEventFlagsId_t ef_id)
{
    return 0;
}

/// Wait for one or more Event Flags to become signaled.
/// \param[in]     ef_id         event flags ID obtained by \ref osEventFlagsNew.
/// \param[in]     flags         specifies the flags to wait for.
/// \param[in]     options       specifies flags options (osFlagsXxxx).
/// \param[in]     timeout       \ref CMSIS_RTOS_TimeOutValue or 0 in case of no time-out.
/// \return event flags before clearing or error code if highest bit set.
static uint32_t osEventFlagsWaitByLinux (osEventFlagsId_t ef_id, uint32_t flags, uint32_t options, uint32_t timeout)
{
    return 0;
}

/// Delete an Event Flags object.
/// \param[in]     ef_id         event flags ID obtained by \ref osEventFlagsNew.
/// \return status code that indicates the execution status of the function.
static osStatus_t osEventFlagsDeleteByLinux (osEventFlagsId_t ef_id)
{
    return osOK;
}


//  ==== Mutex Management Functions ====

/// Create and Initialize a Mutex object.
/// \param[in]     attr          mutex attributes; NULL: default values.
/// \return mutex ID for reference by other functions or NULL in case of error.
static osMutexId_t osMutexNewByLinux (const osMutexAttr_t *attr)
{
    int32_t iRet;
    MUTEX_LINUX_CB_S *pstMutexCb = NULL;

    pstMutexCb = (MUTEX_LINUX_CB_S *)osMemAlloc(sizeof(MUTEX_LINUX_CB_S));
    if (!pstMutexCb)
    {
        return (osMutexId_t)NULL;
    }

    iRet = pthread_mutex_init(&pstMutexCb->stMutex, &pstMutexCb->stMutexAttr);
    if (0 != iRet)
    {
        osMemFree((void **)&pstMutexCb);
        return (osMutexId_t)NULL;
    }

    if (attr)
    {
        pstMutexCb->pName = attr->name;
    }
    pstMutexCb->pOwner = osThreadGetId();
    return (osMutexId_t)pstMutexCb;
}

/// Get name of a Mutex object.
/// \param[in]     mutex_id      mutex ID obtained by \ref osMutexNew.
/// \return name as null-terminated string.
static const char *osMutexGetNameByLinux(osMutexId_t mutex_id)
{
    MUTEX_LINUX_CB_S *pstMutexCb = (MUTEX_LINUX_CB_S *)mutex_id;

    if (!pstMutexCb)
    {
        return "unknow";
    }

    if (!pstMutexCb->pName)
    {
        return "unknow";
    }
    return pstMutexCb->pName;
}

/// Acquire a Mutex or timeout if it is locked.
/// \param[in]     mutex_id      mutex ID obtained by \ref osMutexNew.
/// \param[in]     timeout       \ref CMSIS_RTOS_TimeOutValue or 0 in case of no time-out.
/// \return status code that indicates the execution status of the function.
static osStatus_t osMutexAcquireByLinux(osMutexId_t mutex_id, uint32_t timeout)
{
    int32_t iRet;
    MUTEX_LINUX_CB_S *pstMutexCb = (MUTEX_LINUX_CB_S *)mutex_id;

    if (!pstMutexCb)
    {
        return osError;
    }

    if (0 == timeout)
    {
        iRet = pthread_mutex_lock(&pstMutexCb->stMutex);
    }
    else
    {
        struct timespec stWaitTime = { 0 };
        
        osWaitTimerInitByLinux(&stWaitTime, timeout);
        iRet = pthread_mutex_timedlock(&pstMutexCb->stMutex, &stWaitTime);
    }

    if (0 != iRet)
    {
        return osError;
    }
    return osOK;
}

/// Release a Mutex that was acquired by \ref osMutexAcquire.
/// \param[in]     mutex_id      mutex ID obtained by \ref osMutexNew.
/// \return status code that indicates the execution status of the function.
static osStatus_t osMutexReleaseByLinux (osMutexId_t mutex_id)
{
    MUTEX_LINUX_CB_S *pstMutexCb = (MUTEX_LINUX_CB_S *)mutex_id;

    if (!pstMutexCb)
    {
        return osError;
    }

    if (0 != pthread_mutex_unlock(&pstMutexCb->stMutex))
    {
        return osError;
    }
    return osOK;
}

/// Get Thread which owns a Mutex object.
/// \param[in]     mutex_id      mutex ID obtained by \ref osMutexNew.
/// \return thread ID of owner thread or NULL when mutex was not acquired.
static osThreadId_t osMutexGetOwnerByLinux (osMutexId_t mutex_id)
{
    MUTEX_LINUX_CB_S *pstMutexCb = (MUTEX_LINUX_CB_S *)mutex_id;

    if (!pstMutexCb)
    {
        return (osThreadId_t)NULL;
    }

    return pstMutexCb->pOwner;
}

/// Delete a Mutex object.
/// \param[in]     mutex_id      mutex ID obtained by \ref osMutexNew.
/// \return status code that indicates the execution status of the function.
static osStatus_t osMutexDeleteByLinux (osMutexId_t mutex_id)
{
    MUTEX_LINUX_CB_S *pstMutexCb = (MUTEX_LINUX_CB_S *)mutex_id;

    if (!pstMutexCb)
    {
        return osError;
    }

    if (0 > pthread_mutex_destroy(&pstMutexCb->stMutex))
    {
        return osError;
    }

    osMemFree((void **)&pstMutexCb);
    return osOK;
}


//  ==== Semaphore Management Functions ====

/// Create and Initialize a Semaphore object.
/// \param[in]     max_count     maximum number of available tokens.
/// \param[in]     initial_count initial number of available tokens.
/// \param[in]     attr          semaphore attributes; NULL: default values.
/// \return semaphore ID for reference by other functions or NULL in case of error.
static osSemaphoreId_t osSemaphoreNewByLinux (uint32_t max_count, uint32_t initial_count, const osSemaphoreAttr_t *attr)
{
    SEM_LINUX_CB_S *pstSemCb = NULL;

    pstSemCb = (SEM_LINUX_CB_S *)osMemAlloc(sizeof(SEM_LINUX_CB_S));
    if (!pstSemCb)
    {
        return (osSemaphoreId_t)NULL;
    }

    if (0 != sem_init(&pstSemCb->stSem, 0, initial_count))
    {
        osMemFree((void **)&pstSemCb);
        return (osSemaphoreId_t)NULL;
    }

    if (attr)
    {
        pstSemCb->pName = attr->name;
    }
    return (osSemaphoreId_t)pstSemCb;
}

/// Get name of a Semaphore object.
/// \param[in]     semaphore_id  semaphore ID obtained by \ref osSemaphoreNew.
/// \return name as null-terminated string.
static const char *osSemaphoreGetNameByLinux (osSemaphoreId_t semaphore_id)
{
    SEM_LINUX_CB_S *pstSemCb = (SEM_LINUX_CB_S *)semaphore_id;

    if (!pstSemCb)
    {
        return "unknow";
    }

    if (!pstSemCb->pName)
    {
        return "unknow";
    }

    return pstSemCb->pName;
}

/// Acquire a Semaphore token or timeout if no tokens are available.
/// \param[in]     semaphore_id  semaphore ID obtained by \ref osSemaphoreNew.
/// \param[in]     timeout       \ref CMSIS_RTOS_TimeOutValue or 0 in case of no time-out.
/// \return status code that indicates the execution status of the function.
static osStatus_t osSemaphoreAcquireByLinux (osSemaphoreId_t semaphore_id, uint32_t timeout)
{
    int32_t iRet;
    SEM_LINUX_CB_S *pstSemCb = (SEM_LINUX_CB_S *)semaphore_id;

    if (!pstSemCb)
    {
        return osError;
    }

    // sem_wait or sem_timedwait
    if (!timeout)
    {
        iRet = sem_wait(&pstSemCb->stSem);
    }
    else
    {
        struct timespec stWaitTime = { 0 };
        
        osWaitTimerInitByLinux(&stWaitTime, timeout);
        iRet = sem_timedwait(&pstSemCb->stSem, &stWaitTime);
    }

    if (0 != iRet)
    {
        return osError;
    }

    return osOK;
}

/// Release a Semaphore token up to the initial maximum count.
/// \param[in]     semaphore_id  semaphore ID obtained by \ref osSemaphoreNew.
/// \return status code that indicates the execution status of the function.
static osStatus_t osSemaphoreReleaseByLinux (osSemaphoreId_t semaphore_id)
{
    SEM_LINUX_CB_S *pstSemCb = (SEM_LINUX_CB_S *)semaphore_id;

    if (!pstSemCb)
    {
        return osError;
    }

    if (0 != sem_post(&pstSemCb->stSem))
    {
        return osError;
    }

    return osOK;
}

/// Get current Semaphore token count.
/// \param[in]     semaphore_id  semaphore ID obtained by \ref osSemaphoreNew.
/// \return number of tokens available.
static uint32_t osSemaphoreGetCountByLinux (osSemaphoreId_t semaphore_id)
{
    int32_t uiCount = 0;
    SEM_LINUX_CB_S *pstSemCb = (SEM_LINUX_CB_S *)semaphore_id;

    if (!pstSemCb)
    {
        return 0;
    }

    sem_getvalue(&pstSemCb->stSem, &uiCount);
    return uiCount;
}

/// Delete a Semaphore object.
/// \param[in]     semaphore_id  semaphore ID obtained by \ref osSemaphoreNew.
/// \return status code that indicates the execution status of the function.
static osStatus_t osSemaphoreDeleteByLinux (osSemaphoreId_t semaphore_id)
{
    SEM_LINUX_CB_S *pstSemCb = (SEM_LINUX_CB_S *)semaphore_id;

    if (!pstSemCb)
    {
        return osError;
    }

    if (0 != sem_destroy(&pstSemCb->stSem))
    {
        return osError;
    }

    osMemFree((void **)&pstSemCb);
    return osOK;
}

//  ==== Message Queue Management Functions ====

/// Create and Initialize a Message Queue object.
/// \param[in]     msg_count     maximum number of messages in queue.
/// \param[in]     msg_size      maximum message size in bytes.
/// \param[in]     attr          message queue attributes; NULL: default values.
/// \return message queue ID for reference by other functions or NULL in case of error.
static osMessageQueueId_t osMessageQueueNewByLinux (uint32_t msg_count, uint32_t msg_size, const osMessageQueueAttr_t *attr)
{
    mqd_t mq_cmd;
    char mq_name[32] = { 0 };
    struct mq_attr mq_attr;
    MSG_LINUX_CB_S *pstMsgCb = NULL;

    if (!attr)
    {
        return (osMessageQueueId_t)NULL;
    }

    pstMsgCb = (MSG_LINUX_CB_S *)osMemAlloc(sizeof(MSG_LINUX_CB_S));
    if (!pstMsgCb)
    {
        return (osMessageQueueId_t)NULL;
    }

    if ('/' != attr->name[0])
    {
        sprintf(mq_name, "/%s", attr->name);
    }
    else
    {
        strcpy(mq_name, attr->name);
    }

    mq_attr.mq_flags = 0;
    mq_attr.mq_maxmsg = msg_count;
    mq_attr.mq_msgsize = msg_size;
    mq_attr.mq_curmsgs = 0;
    mq_cmd = mq_open(mq_name, O_CREAT | O_RDWR, 0664, &mq_attr);
    if (0 > mq_cmd)
    {
        osMemFree((void **)&pstMsgCb);
        return (osMessageQueueId_t)NULL;
    }

    pstMsgCb->pName = attr->name;
    pstMsgCb->uiMsgCnt = msg_count;
    pstMsgCb->uiMsgSize = msg_size;
    pstMsgCb->iMsgQCmd = mq_cmd;
    return (osMessageQueueId_t)pstMsgCb;
}

/// Get name of a Message Queue object.
/// \param[in]     mq_id         message queue ID obtained by \ref osMessageQueueNew.
/// \return name as null-terminated string.
static const char *osMessageQueueGetNameByLinux (osMessageQueueId_t mq_id)
{
    MSG_LINUX_CB_S *pstMsgCb = (MSG_LINUX_CB_S *)mq_id;

    if (!pstMsgCb)
    {
        return "unknow";
    }

    if (!pstMsgCb->pName)
    {
        return "unknow";
    }

    return pstMsgCb->pName;
}

/// Put a Message into a Queue or timeout if Queue is full.
/// \param[in]     mq_id         message queue ID obtained by \ref osMessageQueueNew.
/// \param[in]     msg_ptr       pointer to buffer with message to put into a queue.
/// \param[in]     msg_prio      message priority.
/// \param[in]     timeout       \ref CMSIS_RTOS_TimeOutValue or 0 in case of no time-out.
/// \return status code that indicates the execution status of the function.
static osStatus_t osMessageQueuePutByLinux (osMessageQueueId_t mq_id, const void *msg_ptr, uint8_t msg_prio, uint32_t timeout)
{
    int32_t iRet;
    MSG_LINUX_CB_S *pstMsgCb = (MSG_LINUX_CB_S *)mq_id;

    if (!pstMsgCb)
    {
        return osError;
    }

    if (0 == timeout)
    {
        iRet = mq_send(pstMsgCb->iMsgQCmd, msg_ptr, pstMsgCb->uiMsgSize, msg_prio);
    }
    else
    {
        struct timespec stWaitTime = { 0 };
        
        osWaitTimerInitByLinux(&stWaitTime, timeout);
        iRet = mq_timedsend(pstMsgCb->iMsgQCmd, msg_ptr, pstMsgCb->uiMsgSize, msg_prio, &stWaitTime);
    }

    if (0 != iRet)
    {
        return osError;
    }
    return osOK;
}

/// Get a Message from a Queue or timeout if Queue is empty.
/// \param[in]     mq_id         message queue ID obtained by \ref osMessageQueueNew.
/// \param[out]    msg_ptr       pointer to buffer for message to get from a queue.
/// \param[out]    msg_prio      pointer to buffer for message priority or NULL.
/// \param[in]     timeout       \ref CMSIS_RTOS_TimeOutValue or 0 in case of no time-out.
/// \return status code that indicates the execution status of the function.
static osStatus_t osMessageQueueGetByLinux (osMessageQueueId_t mq_id, void *msg_ptr, uint8_t *msg_prio, uint32_t timeout)
{
    int32_t iRet;
    uint32_t uiMsgPrio = 0;
    MSG_LINUX_CB_S *pstMsgCb = (MSG_LINUX_CB_S *)mq_id;

    if (!pstMsgCb)
    {
        return osError;
    }

    if (0 == timeout)
    {
        iRet = mq_receive(pstMsgCb->iMsgQCmd, msg_ptr, pstMsgCb->uiMsgSize, &uiMsgPrio);
    }
    else
    {
        struct timespec stWaitTime = { 0 };
        
        osWaitTimerInitByLinux(&stWaitTime, timeout);
        iRet = mq_timedreceive(pstMsgCb->iMsgQCmd, msg_ptr, pstMsgCb->uiMsgSize, &uiMsgPrio, &stWaitTime);
    }

    if (0 != iRet)
    {
        return osError;
    }

    if (msg_prio)
    {
        *msg_prio = (uint8_t)uiMsgPrio;
    }
    return osOK;
}

/// Get maximum number of messages in a Message Queue.
/// \param[in]     mq_id         message queue ID obtained by \ref osMessageQueueNew.
/// \return maximum number of messages.
static uint32_t osMessageQueueGetCapacityByLinux (osMessageQueueId_t mq_id)
{
    MSG_LINUX_CB_S *pstMsgCb = (MSG_LINUX_CB_S *)mq_id;

    if (!pstMsgCb)
    {
        return 0;
    }
    return pstMsgCb->uiMsgCnt;
}

/// Get maximum message size in a Memory Pool.
/// \param[in]     mq_id         message queue ID obtained by \ref osMessageQueueNew.
/// \return maximum message size in bytes.
static uint32_t osMessageQueueGetMsgSizeByLinux (osMessageQueueId_t mq_id)
{
    MSG_LINUX_CB_S *pstMsgCb = (MSG_LINUX_CB_S *)mq_id;

    if (!pstMsgCb)
    {
        return 0;
    }
    return pstMsgCb->uiMsgSize;
}

/// Get number of queued messages in a Message Queue.
/// \param[in]     mq_id         message queue ID obtained by \ref osMessageQueueNew.
/// \return number of queued messages.
static uint32_t osMessageQueueGetCountByLinux (osMessageQueueId_t mq_id)
{
    struct mq_attr mqAttr;
    MSG_LINUX_CB_S *pstMsgCb = (MSG_LINUX_CB_S *)mq_id;

    if (!pstMsgCb)
    {
        return 0;
    }

    if (mq_getattr(pstMsgCb->iMsgQCmd, &mqAttr) < 0)
    {
        return 0;
    }

    return mqAttr.mq_curmsgs;
}

/// Get number of available slots for messages in a Message Queue.
/// \param[in]     mq_id         message queue ID obtained by \ref osMessageQueueNew.
/// \return number of available slots for messages.
static uint32_t osMessageQueueGetSpaceByLinux (osMessageQueueId_t mq_id)
{
    uint32_t uiMaxCnt = osMessageQueueGetCapacityByLinux(mq_id);
    uint32_t uiCurCnt = osMessageQueueGetCountByLinux(mq_id);
    
    return (uiMaxCnt > uiCurCnt)?(uiMaxCnt - uiCurCnt):0;
}

/// Reset a Message Queue to initial empty state.
/// \param[in]     mq_id         message queue ID obtained by \ref osMessageQueueNew.
/// \return status code that indicates the execution status of the function.
static osStatus_t osMessageQueueResetByLinux (osMessageQueueId_t mq_id)
{
    return osOK;
}

/// Delete a Message Queue object.
/// \param[in]     mq_id         message queue ID obtained by \ref osMessageQueueNew.
/// \return status code that indicates the execution status of the function.
static osStatus_t osMessageQueueDeleteByLinux (osMessageQueueId_t mq_id)
{
    char mq_name[32] = { 0 };
    MSG_LINUX_CB_S *pstMsgCb = (MSG_LINUX_CB_S *)mq_id;

    if (!pstMsgCb)
    {
        return osError;
    }

    if ('/' != pstMsgCb->pName[0])
    {
        sprintf(mq_name, "/%s", pstMsgCb->pName);
    }
    else
    {
        strcpy(mq_name, pstMsgCb->pName);
    }

    mq_close(pstMsgCb->iMsgQCmd);
    mq_unlink(mq_name);
    return osOK;
}

/**
*   @Description: 时间转换
*   @param pOutTime - 出参，TimedWait需要的时间格式，当前时间+等待时间
*   @param uiMs - 入参，需要等待的时间，单位毫秒
*   @return: NA
*/
static void osWaitTimerInitByLinux(struct timespec *pOutTime, uint32_t uiMs)
{
    struct timeval stCurTime;

    gettimeofday(&stCurTime, NULL);

    pOutTime->tv_sec = stCurTime.tv_sec;
    pOutTime->tv_nsec = stCurTime.tv_usec * 1000;

    pOutTime->tv_sec += uiMs / 1000;
    pOutTime->tv_nsec += (uiMs % 1000) * 1000 * 1000;

    pOutTime->tv_sec += pOutTime->tv_nsec / 1000 / 1000 / 1000;
    pOutTime->tv_nsec %= 1000 * 1000 * 1000;
}

#endif  // SAL_PRODUCER == _SAL_LINUX_

/************************************** The End Of File **************************************/



